package cn.com.bluemoon.daps.model.service.impl;


import cn.com.bluemoon.daps.api.notice.RemoteNoticeHandlerService;
import cn.com.bluemoon.daps.api.sync.RemoteSyncService;
import cn.com.bluemoon.daps.api.sys.RemoteSystemDatasourceService;
import cn.com.bluemoon.daps.api.sys.RemoteSystemUserService;
import cn.com.bluemoon.daps.common.domain.BaseModel;
import cn.com.bluemoon.daps.common.domain.ResultBean;
import cn.com.bluemoon.daps.common.domain.UserInfoHolder;
import cn.com.bluemoon.daps.common.enums.BmStatus;
import cn.com.bluemoon.daps.common.enums.DatabaseType;
import cn.com.bluemoon.daps.common.exception.DapException;
import cn.com.bluemoon.daps.common.toolkit.JsonUtil;
import cn.com.bluemoon.daps.model.dto.DapDataModelTableDto;
import cn.com.bluemoon.daps.model.dto.score.TableScoreDto;
import cn.com.bluemoon.daps.model.entity.*;
import cn.com.bluemoon.daps.model.mapper.DapDataModelRecordMapper;
import cn.com.bluemoon.daps.model.service.*;
import cn.com.bluemoon.daps.model.utils.contrast.VersionContrast;
import cn.com.bluemoon.daps.model.utils.email.EmailUtils;
import cn.com.bluemoon.daps.model.vo.SearchModelApplicationVo;
import cn.com.bluemoon.daps.model.vo.SearchModelVersionVo;
import cn.com.bluemoon.daps.notice.core.NoticeType;
import cn.com.bluemoon.daps.notice.template.ApplyPublishMessage;
import cn.com.bluemoon.daps.standard.dto.ApplicationModelDto;
import cn.com.bluemoon.daps.sync.base.ModifyType;
import cn.com.bluemoon.daps.sync.vo.applypublish.ApplyPublishParam;
import cn.com.bluemoon.daps.sync.vo.applypublish.ModelCompareVo;
import cn.com.bluemoon.daps.system.dto.SystemUserDto;
import cn.com.bluemoon.daps.system.entity.DapSystemDatasource;
import cn.com.bluemoon.metadata.inter.RemoteMetaDataTaskJobService;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;

/**
 * <p>
 * 申请发布审批状态表 服务实现类
 * </p>
 *
 * @author admin
 * @since 2020-08-03
 */
@Slf4j
@Service
public class DapDataModelRecordServiceImpl extends ServiceImpl<DapDataModelRecordMapper, DapDataModelRecord> implements DapDataModelRecordService {

    @Resource
    DapDataModelTableService dapDataModelTableService;

    @Resource
    DapDataModelTableFieldService dapDataModelTableFieldService;

    @Resource
    DapDataModelSubordinateKeyService dapDataModelSubordinateKeyService;

    @Resource
    DapDataModelTablePartitionService dapDataModelTablePartitionService;

    @Resource
    DapDataModelOperateHistoryService dapDataModelOperateHistoryService;

    @Resource
    DapDataModelRecordMapper modelRecordMapper;

    @Resource
    DapDataModelVersionService dapDataModelVersionService;

    @Resource
    RemoteSyncService remoteSyncService;

    @Resource
    DapDataModelService dapDataModelService;

    @Resource
    RemoteSystemDatasourceService remoteSystemDatasourceService;

    @Resource
    DapDataModelTableScoreRuleService dapDataModelTableScoreRuleService;

    @Resource
    RemoteNoticeHandlerService remoteNoticeHandlerService;

    @Resource
    RemoteSystemUserService remoteSystemUserService;

    @Resource
    RemoteMetaDataTaskJobService remoteMetaDataTaskJobService;

    /**
     * 申请列表
     *
     * @param application
     * @return
     */
    @Override
    public IPage<ApplicationModelDto> getList(SearchModelApplicationVo application, Integer applicationType) {

        Page<ApplicationModelDto> pages = modelRecordMapper.selectApplicationList(new Page<>(application.getPageNum(), application.getPageSize()), application, applicationType);
        return pages;
    }

    /**
     * 单个模型申请发布
     *
     * @param recordId
     */
    @Override
    @Transactional(rollbackFor = {DapException.class, Exception.class})
    public ResultBean<Boolean> singleModelApplicationRelease(String recordId, String desc) {

        // 查询当前用户是否有发布的权限
        DapDataModelRecord dapDataModelRecord = this.getBaseMapper().selectById(recordId);
        DapDataModel dapDataModel = dapDataModelService.getBaseMapper().selectById(dapDataModelRecord.getModelId());
        String account = UserInfoHolder.getAccount();
        if(!account.equals(dapDataModel.getModelAdmin())){
            return ResultBean.error("只允许模型管理员进行发布");
        }
        // 首先将状态改为待审批
        boolean update = this.update(new LambdaUpdateWrapper<DapDataModelRecord>()
                .set(DapDataModelRecord::getPublicStatus, 3)
                .set(DapDataModelRecord::getApplicationTime, LocalDateTime.now())
                .set(DapDataModelRecord::getApplicationBy, UserInfoHolder.getAccount())
                .set(DapDataModelRecord::getChangeDesc, desc)
                .eq(DapDataModelRecord::getId, recordId));
        // 高危语句、自动审核、表结构评分
        ResultBean<DapSystemDatasource> datasource = remoteSystemDatasourceService.getDatasource(dapDataModel.getDataSourceId());
        if(datasource.getContent().getDatabaseType().equals(DatabaseType.MYSQL)){
            String modelId = dapDataModel.getId();
            DatabaseType databaseType = datasource.getContent().getDatabaseType();
            // 是否存在高危语句，查询删除和修改表的状态
            List<String> dropSql = getDropSql(recordId);
            // 是否自动审核 1 通过、2.不通过、3未开启
            Integer autoPass = datasource.getContent().getAutoCheck() ? 2 : 3;
            // 自动审核是否开启
            if(datasource.getContent().getAutoCheck()){
                // 查询模型下属表字段
                List<DapDataModelTableDto> collect = dapDataModelTableService.getCopyTableList(modelId,1);
                // 获得索引或分区信息
                List<DapDataModelTableDto> tableList = getKeyAndPartitionList(collect, databaseType, modelId, true);
                List<TableScoreDto> tableScore = dapDataModelTableScoreRuleService.getTableScore(tableList, null); // 评分关卡
                Boolean tableScoreFlag = tableScore.stream().allMatch(s -> s.getTableScore() >= 60);
                autoPass = dropSql.isEmpty() && tableScoreFlag ? 1 : 2;
            }
            // 人工审核 1,2,3,4代表通过、不需要、待审核、不通过
            Integer manAuto = autoPass == 1 ? 2 : 3;
            // 直接通过发布
            if(autoPass == 1){
                this.pass(modelId,recordId,databaseType,false);
            }
            dapDataModelRecord.setChangeDesc(desc);
            // 发送申请人邮件格式
            ApplyPublishMessage message = EmailUtils.sendToUserMessage(autoPass, dapDataModel, dropSql, datasource.getContent()
                    , dapDataModelRecord, manAuto);
            // 发送给用户
            remoteNoticeHandlerService.sendApplyMsg(NoticeType.EMAIL, message);
            // 获取dba列表
            ResultBean<List<SystemUserDto>> listResultBean = remoteSystemUserService.userDba();
            List<String> collect = listResultBean.getContent().stream().map(f -> f.getEmail()).collect(Collectors.toList());
            // 发送dba邮件格式
            ApplyPublishMessage dbaMessage = EmailUtils.sendToDbaMessage(autoPass, UserInfoHolder.getUserName(), dapDataModel, dropSql,
                    datasource.getContent(), dapDataModelRecord, manAuto, collect);
            // 发送给dba
            return remoteNoticeHandlerService.sendApplyMsg(NoticeType.EMAIL, dbaMessage);
        }
        return ResultBean.ok(update);
    }

    /**
     * 查询是否存在高危语句
     * @param recordId
     * @return
     */
    private List<String> getDropSql(String recordId){
        // 是否存在高危语句，查询删除和修改表的状态
        List<DapDataModelOperateHistory> histories = dapDataModelOperateHistoryService.getBaseMapper().selectList(new LambdaQueryWrapper<DapDataModelOperateHistory>()
                .in(DapDataModelOperateHistory::getLastChangeType, 2, 3).eq(BaseModel::getBmStatus, BmStatus.ON.getCode())
                .eq(DapDataModelOperateHistory::getDataModelRecordId, recordId));
        // 筛选出删除表的语句
        List<String> dropSql = histories.stream().filter(h -> h.getLastChangeType() == 2)
                .map(DapDataModelOperateHistory::getPublicContent).collect(Collectors.toList());
        // 修改字段的语句
        List<String> modifyField = histories.stream().filter(h -> h.getLastChangeType() == 3).map(DapDataModelOperateHistory::getPublicContent).collect(Collectors.toList());
        // 解析内容获取删除数据
        for (String content : modifyField){
            Map<String, String> map = JsonUtil.jsonStringToMap(content);
            for (Map.Entry<String, String> entry : map.entrySet()) {
                // 删除字段语句
                if(entry.getKey().equals(ModifyType.DROP.getDesc() + ModifyType.ModifyLevel.FIELD.getDesc())){
                    dropSql.add(entry.getValue());
                }
                // 删除索引语句
                if(entry.getKey().equals(ModifyType.DROP.getDesc() + ModifyType.ModifyLevel.INDEX.getDesc())){
                    dropSql.add(entry.getValue());
                }
            }
        }
        return dropSql;
    }

    /**
     * 撤销
     *
     * @param modelId
     * @param recordId
     */
    @Override
    @Transactional(rollbackFor = {DapException.class, Exception.class})
    public ResultBean<Boolean> cancelRelease(String modelId, String recordId) {

        // 权限控制非模型管理员不能进行撤销
        DapDataModel dapDataModel = dapDataModelService.getBaseMapper().selectById(modelId);
        ResultBean<DapSystemDatasource> datasource = remoteSystemDatasourceService.getDatasource(dapDataModel.getDataSourceId());
        String account = UserInfoHolder.getAccount();
        if(!account.equals(dapDataModel.getModelAdmin())){
            return ResultBean.error("只允许模型管理员进行撤销");
        }
        if(datasource.getCode() == 200 && datasource.getContent().getDatabaseType().equals(DatabaseType.MYSQL)){
            // 获得当前的申请记录
            DapDataModelRecord dapDataModelRecord = this.getBaseMapper().selectById(recordId);
            // 调用同步模块接口
            ApplyPublishParam applyPublishParam = new ApplyPublishParam();
            BeanUtils.copyProperties(dapDataModelRecord, applyPublishParam);
            applyPublishParam.setVerifyBy(UserInfoHolder.getAccount());
            applyPublishParam.setAutoPass(false);
            applyPublishParam.setCompareVersion(dapDataModelRecord.getVersionKey());
            ResultBean<String> result = remoteSyncService.applyRevert(applyPublishParam);
            if(result.getCode() == 500){
                return ResultBean.error(result.getMsg());
            }
        }
        // 首先删除此条记录
        modelRecordMapper.deleteById(recordId);
        // 首先判断是否为第一次进行发布
        List<DapDataModelRecord> dapDataModelRecords = this.getRecord(modelId);
        // 为空表示第一次进行发布
        if (dapDataModelRecords.isEmpty()) {
            // 删除复制的表
            dapDataModelTableService.deleteTable(modelId);
            // 删除备份的字段
            dapDataModelTableFieldService.deleteField(modelId);
            // 删除备份的索引
            dapDataModelSubordinateKeyService.deleteKey(modelId);
            // 删除备份的分区
            dapDataModelTablePartitionService.deletePartition(modelId);
        }
        dapDataModelOperateHistoryService.getBaseMapper().delete(new LambdaQueryWrapper<DapDataModelOperateHistory>()
                .eq(DapDataModelOperateHistory::getDataModelRecordId, recordId));
        return ResultBean.ok(true);
    }

    /**
     * 不通过发布
     *
     * @param recordId
     */
    @Override
    public void unPass(String recordId) {

        this.update(new LambdaUpdateWrapper<DapDataModelRecord>()
                .set(DapDataModelRecord::getPublicStatus, 4)
                .set(DapDataModelRecord::getVerifyBy, UserInfoHolder.getAccount())
                .eq(DapDataModelRecord::getId, recordId));
        // 查询记录
        DapDataModelRecord record = this.getBaseMapper().selectById(recordId);
        // 获得模型
        DapDataModel dapDataModel = dapDataModelService.getBaseMapper().selectById(record.getModelId());
        // 获得数据源
        ResultBean<DapSystemDatasource> datasource = remoteSystemDatasourceService.getDatasource(dapDataModel.getDataSourceId());
        // 申请人邮件格式
        if(datasource.getContent().getDatabaseType().equals(DatabaseType.MYSQL)){
            ApplyPublishMessage message = EmailUtils.sendToUserMessage(datasource.getContent().getAutoCheck() ? 2 : 3, dapDataModel, getDropSql(recordId), datasource.getContent()
                    , record, 4);
            // 发送给用户
            remoteNoticeHandlerService.sendApplyMsg(NoticeType.EMAIL, message);
        }
        // 同步模块记录不通过操作
        ApplyPublishParam applyPublishParam = new ApplyPublishParam();
        BeanUtils.copyProperties(record, applyPublishParam);
        applyPublishParam.setVerifyBy(UserInfoHolder.getAccount());
        applyPublishParam.setAutoPass(false);
        applyPublishParam.setCompareVersion(record.getVersionKey());
        remoteSyncService.applyConfirmUnPass(applyPublishParam);
    }

    /**
     * 通过发布
     *
     * @param recordId
     */
    @Override
    @Transactional(rollbackFor = {DapException.class, Exception.class})
    public ResultBean<String> pass(String modelId, String recordId, DatabaseType databaseType, Boolean isMenAuto) {

        // 如果是第一次进行发布
        List<DapDataModelRecord> dapDataModelRecords = getRecord(modelId);
        List<DapDataModelRecord> records = dapDataModelRecords.stream()
                .filter(r -> r.getId().equals(recordId) && r.getPublicStatus() == 3).collect(Collectors.toList());
        if(records.isEmpty()){
            throw new DapException("发布单号不是待审批状态，无法进行发布");
        }
        // 存在发布版本
        if (dapDataModelRecords.size() > 1) {
            // *最新发布时间
            LocalDateTime max = dapDataModelRecords.stream().filter(d -> d.getReleaseTime() != null)
                    .map(DapDataModelRecord::getReleaseTime).distinct().max((e1, e2) -> e1.compareTo(e2)).get();
            // *查询出更新时间大于发布时间的表
            List<DapDataModelTable> newTable = dapDataModelTableService.getBaseMapper().selectList(new LambdaQueryWrapper<DapDataModelTable>()
                    .eq(DapDataModelTable::getDataModelId, modelId).gt(DapDataModelTable::getUpdateTime, max)
                    .eq(BaseModel::getBmStatus, BmStatus.ON.getCode()).eq(DapDataModelTable::getPId, "0"));
            // 查询模型发布记录
            List<DapDataModelOperateHistory> histories = dapDataModelOperateHistoryService.getBaseMapper().selectList(new LambdaQueryWrapper<DapDataModelOperateHistory>()
                    .eq(DapDataModelOperateHistory::getDataModelRecordId, recordId)
                    .eq(DapDataModelOperateHistory::getBmStatus, BmStatus.ON.getCode()));
            // 新增和变化的表id
            List<String> addAndChangeIds = histories.stream().filter(h -> h.getLastChangeType() ==  1 || h.getLastChangeType() == 3)
                    .map(DapDataModelOperateHistory::getModelTableId).collect(Collectors.toList());
            // 删除和变化的表id
            List<String> deleteAndChangeIds = histories.stream().filter(h -> h.getLastChangeType() == 2 || h.getLastChangeType() == 3)
                    .map(DapDataModelOperateHistory::getModelTableId).collect(Collectors.toList());
            // *过滤出未发生表结构变更但是表其他信息发生变化
            List<String> ids = newTable.stream().filter(n -> !addAndChangeIds.contains(n.getId())).map(DapDataModelTable::getId).collect(Collectors.toList());
            // *加入新增表和删除表的队列
            if(!ids.isEmpty()){
                addAndChangeIds.addAll(ids);deleteAndChangeIds.addAll(ids);
            }
            // 查询模型新版本数据
            List<DapDataModelTableDto> tableList = dapDataModelTableService.getCopyTableList(modelId,1);
            List<DapDataModelTableDto> newList = getKeyAndPartitionList(tableList, databaseType, modelId, true);
            // 找出新增和变化的新表
            List<DapDataModelTableDto> addTables = newList.stream().filter(n -> addAndChangeIds.contains(n.getTableId())).collect(Collectors.toList());
            // 查询模型旧版本数据
            List<DapDataModelTableDto> tables = dapDataModelTableService.getCopyTableList(modelId, 2);
            List<DapDataModelTableDto> oldList = getKeyAndPartitionList(tables, databaseType, modelId, false);
            // 找出删除和变化的旧表
            List<DapDataModelTableDto> deleteTables = oldList.stream().filter(o -> deleteAndChangeIds.contains(o.getPId())).collect(Collectors.toList());
           // 版本记录删除旧表
            if(!deleteTables.isEmpty()){
                List<String> deleteAndModifyIds = deleteTables.stream().map(DapDataModelTableDto::getTableId).collect(Collectors.toList());
                List<String> fieldIds = new ArrayList<>();
                List<String> keyIds = new ArrayList<>();
                deleteTables.forEach(d -> {
                    List<String> collect = d.getFieldList().stream().map(DapDataModelTableField::getId).collect(Collectors.toList());
                    fieldIds.addAll(collect);
                    List<String> keys = d.getKeyList().stream().map(DapDataModelSubordinateKey::getId).collect(Collectors.toList());
                    if(!keys.isEmpty()){
                        keyIds.addAll(keys);
                    }
                });
                // 删除表
                dapDataModelTableService.getBaseMapper().deleteBatchIds(deleteAndModifyIds);
                // 删除字段
                dapDataModelTableFieldService.getBaseMapper().deleteBatchIds(fieldIds);
                // 删除索引
                if(!keyIds.isEmpty()){
                    dapDataModelTableFieldService.getBaseMapper().deleteBatchIds(keyIds);
                }
            }
            // 重新生成发布数据
            if(!addTables.isEmpty()){
                insertVersionData(addTables, databaseType);
            }
        }
        // 将状态置为已发布,更新发布时间
        modelRecordMapper.updatePublicTime(recordId, UserInfoHolder.getAccount());
        // 将发布数据的状态置为已发布，更新表、字段、索引、分区表中的状态为发布状态
        modelRecordMapper.updateApplicationStatus(modelId);
        // 查询最新已发布的数据
        List<DapDataModelTableDto> tables = dapDataModelTableService.getCopyTableList(modelId, 2);
        List<DapDataModelTableDto> oldList = getKeyAndPartitionList(tables, databaseType, modelId, false);
        // 转成json字符串
        List<DapDataModelVersion> versions = Lists.newArrayList();
        if(!oldList.isEmpty()){
            oldList.forEach(t ->{
                DapDataModelVersion dapDataModelVersion = new DapDataModelVersion();
                dapDataModelVersion.setModelId(t.getDataModelId());
                dapDataModelVersion.setModelTableId(t.getPId());
                dapDataModelVersion.setPublishStatus(1);
                String string = JSONObject.toJSONString(t);
                dapDataModelVersion.setVersionContent(string);
                dapDataModelVersion.setRecordId(recordId);
                versions.add(dapDataModelVersion);
            });
        }
        // 批量入版本库
        dapDataModelVersionService.saveBatch(versions);
        // 获取模型
        DapDataModel dapDataModel = dapDataModelService.getBaseMapper().selectById(modelId);
        if(databaseType.equals(DatabaseType.MYSQL)){
            // 获得当前的申请记录，如果不需要同步则调用applyRevert接口，需要同步调用通过applyConfirmPass接口
            DapDataModelRecord dapDataModelRecord = this.getBaseMapper().selectById(recordId);
            // 调用同步模块接口
            ApplyPublishParam applyPublishParam = new ApplyPublishParam();
            BeanUtils.copyProperties(dapDataModelRecord, applyPublishParam);
            applyPublishParam.setVerifyBy(UserInfoHolder.getAccount());
            applyPublishParam.setCompareVersion(dapDataModelRecord.getVersionKey());
            applyPublishParam.setAutoPass(!isMenAuto);
            if(dapDataModel.getIsSyncProd()){
                remoteSyncService.applyConfirmPass(applyPublishParam);
            }else {
                remoteSyncService.applyRevert(applyPublishParam);
            }

            // 发送邮件
            if(isMenAuto){
                // 获取当前记录
                DapDataModelRecord record = dapDataModelRecords.stream().filter(d -> d.getId().equals(recordId)).findFirst().get();
                // 获取数据源
                ResultBean<DapSystemDatasource> datasource = remoteSystemDatasourceService.getDatasource(dapDataModel.getDataSourceId());
                // 查询是否存在高危语句
                ApplyPublishMessage message = EmailUtils.sendToUserMessage(datasource.getContent().getAutoCheck() ? 2 : 3, dapDataModel, getDropSql(recordId)
                        , datasource.getContent(), record, 1);
                // 发送给用户
                remoteNoticeHandlerService.sendApplyMsg(NoticeType.EMAIL, message);
            }
        }
        // 调用元数据模块
        cn.com.bluemoon.metadata.common.ResultBean<String> resultBean = remoteMetaDataTaskJobService.syncModelMetaData(dapDataModel.getSystemId());
        return ResultBean.ok(resultBean.getCode() == 500 ? resultBean.getMsg().replace("DescribeException:","") : null);
    }

    @Override
    @Transactional(rollbackFor = {DapException.class, Exception.class})
    public List<DapDataModelRecord> getRecord(String modelId){

        List<DapDataModelRecord> dapDataModelRecords = this.getBaseMapper()
                .selectList(new LambdaQueryWrapper<DapDataModelRecord>()
                        .eq(DapDataModelRecord::getModelId, modelId)
                        .eq(DapDataModelRecord::getBmStatus, BmStatus.ON.getCode()));
        return dapDataModelRecords;
    }

    /**
     * 处于审批流程的记录
     * @param modelId
     * @return
     */
    @Override
    public boolean getUnRelease(String modelId) {

        DapDataModelRecord record = getBaseMapper()
                .selectOne(new LambdaQueryWrapper<DapDataModelRecord>()
                        .isNull(DapDataModelRecord::getReleaseTime)
                        .eq(DapDataModelRecord::getModelId, modelId)
                        .eq(DapDataModelRecord::getBmStatus, BmStatus.ON.getCode()));
        return record == null ? true : false;
    }

    @Override
    public Optional<LocalDateTime> findMaxReleaseTimeByModelId(String modelId) {
        if (modelId != null) {
            LambdaQueryWrapper<DapDataModelRecord> topOneWrapper = bmLambdaQuery().eq(DapDataModelRecord::getModelId, modelId)
                    .select(DapDataModelRecord::getApplicationTime)
                    .orderByDesc(DapDataModelRecord::getApplicationTime).last("limit 1");
            Optional<DapDataModelRecord> maxReleaseModelOpt = Optional.ofNullable(getOne(topOneWrapper));
            if (maxReleaseModelOpt.isPresent()){
                return Optional.ofNullable(maxReleaseModelOpt.get().getApplicationTime());
            }
        }
        return Optional.empty();
    }

    public ResultBean<Boolean> publicRequire(List<DapDataModelTableDto> tableList){

        // 模型为空不允许发布
        if (tableList.isEmpty()) {
            return ResultBean.error("模型中的内容为空，无法发布！");
        } else {
            // 判断表中的字段是否为空
            String tableNames = tableList.stream().filter(t -> t.getFieldList().stream().findFirst().get().getId() == null)
                    .map(DapDataModelTableDto::getEngName).collect(Collectors.joining(","));
            if(StringUtils.isNotBlank(tableNames)){
                return ResultBean.error(String.format("模型中表%s的内容为空，无法发布！", tableNames));
            }
            return ResultBean.ok(true);
        }
    }
    /**
     * v1.2
     * @param modelId
     * @param databaseType
     * @return
     */
    @Override
    @Transactional(rollbackFor = {DapException.class, Exception.class})
    public ResultBean<Boolean> publishPgOrHiveModel(String modelId, DatabaseType databaseType, String modelAdmin) {

        // 查询模型下属表字段
        List<DapDataModelTableDto> tableList = dapDataModelTableService.getCopyTableList(modelId,1);
        // 发布的条件
        ResultBean<Boolean> resultBean = publicRequire(tableList);
        if (resultBean.getCode() == 500){
            return resultBean;
        }
        List<DapDataModelTableDto> newList = getKeyAndPartitionList(tableList, databaseType, modelId, true);
        // 查询模型中是否存在发布记录
        List<DapDataModelRecord> dapDataModelRecords = getRecord(modelId);
        // 第一次进行发布申请
        if(dapDataModelRecords.isEmpty()){
            // 插入版本数据
            Boolean aBoolean = insertVersionData(tableList, databaseType);
            if (!aBoolean) {
                return ResultBean.error("插入版本数据失败");
            }
            // 插入审批记录数据
            DapDataModelRecord dapDataModelRecord = VersionContrast.entity(modelId);
            boolean save = this.save(dapDataModelRecord);
            // 第一次进行发布，最后变更状态应该是新增表, 插入变更记录
            List<DapDataModelOperateHistory> histories = new ArrayList<>();
            for (DapDataModelTableDto dto : newList) {
                DapDataModelOperateHistory history = VersionContrast.historyList(dto, 1, databaseType);
                history.setDataModelRecordId(dapDataModelRecord.getId());
                histories.add(history);
            }
            dapDataModelOperateHistoryService.saveBatch(histories);
            return ResultBean.ok(save);
        }else{
            // 如果表的更新时间大于模型的审核通过时间，表示进行了修改
            LocalDateTime max = dapDataModelRecords.stream().filter(o -> o.getReleaseTime() != null)
                    .map(DapDataModelRecord::getReleaseTime).distinct().max((e1, e2) -> e1.compareTo(e2)).get();
            // 存在已发布数据，但是不存在审核数据插入数据
            DapDataModelRecord dapDataModelRecord = VersionContrast.entity(modelId);
            // 查询上版本的数据
            List<DapDataModelTableDto> tables = dapDataModelTableService.getCopyTableList(modelId, 2);
            List<DapDataModelTableDto> oldList = getKeyAndPartitionList(tables, databaseType, modelId, false);
            ResultBean<List<DapDataModelOperateHistory>> result = VersionContrast.modelContrast(newList, databaseType, oldList, max, false);
            if (result.getCode() == 200) {
                // 插入修改的sql数据
                dapDataModelOperateHistoryService.saveBatch(result.getContent());
                boolean save = save(dapDataModelRecord);
                dapDataModelOperateHistoryService.updateRecordId(dapDataModelRecord.getId(), modelId);
                return ResultBean.ok(save);
            } else {
                return ResultBean.error(result.getMsg());
            }
        }
    }

    @Override
    public ResultBean<Boolean> publishMysqlModel(DapDataModel dapDataModel, DapSystemDatasource dapSystemDatasource) {

        DatabaseType databaseType = dapSystemDatasource.getDatabaseType();
        String modelId = dapDataModel.getId();
        // 查询模型下属表字段
        List<DapDataModelTableDto> collect = dapDataModelTableService.getCopyTableList(modelId,1);
        // 发布的条件
        ResultBean<Boolean> resultBean = publicRequire(collect);
        if (resultBean.getCode() == 500){
            return resultBean;
        }
        // 获得索引或分区信息
        List<DapDataModelTableDto> tableList = getKeyAndPartitionList(collect, databaseType, modelId, true);
        // 发布同步是否存在未同步的数据，会判断是否存在已发布且未同步的数据
        ResultBean<List<ModelCompareVo>> listResultBean = remoteSyncService.compareModelVersion(modelId, true, false);
        // 对比出结果
        if(listResultBean.getCode() == 500){
            return ResultBean.error(listResultBean.getMsg());
        }
        // 获得对比结果
        List<ModelCompareVo> content = listResultBean.getContent();
        // 查询模型中是否存在发布记录
        List<DapDataModelRecord> dapDataModelRecords = getRecord(modelId);
        // 首次进行发布申请
        if(dapDataModelRecords.isEmpty()) {
            // 版本数据入库
            Boolean flag = insertVersionData(tableList, databaseType);
            if (!flag) {
                return ResultBean.error("插入版本数据失败");
            }
        }
        // 生成审批记录数据
        DapDataModelRecord dapDataModelRecord = VersionContrast.entity(modelId);
        // 版本唯一标识
        dapDataModelRecord.setVersionKey(listResultBean.getMoreData().get("DapSyncProdEnvTask#Id").toString());
        // 审批记录数据入库
        boolean save = this.save(dapDataModelRecord);
        // 解析对比结果
        if(save){
            List<String> ids = content.stream().map(ModelCompareVo::getTableId).collect(Collectors.toList());
            List<DapDataModelTable> tables = dapDataModelTableService.getBaseMapper().selectBatchIds(ids);
            List<DapDataModelOperateHistory> histories = VersionContrast.sqlScriptToData(content, dapDataModelRecord.getId(), tables);
            return ResultBean.ok(dapDataModelOperateHistoryService.saveBatch(histories));
        }else {
            return ResultBean.error("审批记录入库失败，请重新发起");
        }
    }


    /**
     * 生成版本草稿数据
     * @param list
     * @return
     */
    public Boolean insertVersionData(List<DapDataModelTableDto> list, DatabaseType databaseType){

        // 生成表版本数据
        List<DapDataModelTable> tables = new ArrayList<>();
        List<DapDataModelTableField> fields = new ArrayList<>();
        List<DapDataModelSubordinateKey> keys = new ArrayList<>();
        List<DapDataModelTablePartition> partitions = new ArrayList<>();
        for(DapDataModelTableDto dto : list){
            // 表版本数据
            DapDataModelTable table = new DapDataModelTable();
            BeanUtils.copyProperties(dto, table);
            table.setPId(dto.getTableId());
            tables.add(table);
        }
        // 保存表版本数据
        dapDataModelTableService.saveBatch(tables);
        for(DapDataModelTable table : tables){
            for (DapDataModelTableDto dto : list){
                if(dto.getTableId().equals(table.getPId())){
                    // 新版本字段
                    List<DapDataModelTableField> collect = dto.getFieldList().stream()
                            .peek(f -> { f.setModelTableId(table.getId());f.setPId(f.getId());f.setId("");})
                            .collect(Collectors.toList());
                    fields.addAll(collect);
                    // 新版本分区
                    if(null != dto.getPartitionList()){
                        List<DapDataModelTablePartition> partitionList = dto.getPartitionList().stream()
                                .peek(p -> {p.setModelTableId(table.getId());p.setPId(p.getId());p.setId("");})
                                .collect(Collectors.toList());
                        partitions.addAll(partitionList);
                    }
                    // 新版本索引
                    if(null != dto.getKeyList()){
                        List<DapDataModelSubordinateKey> keyList = dto.getKeyList().stream()
                                .peek(k -> {k.setModelTableId(table.getId());k.setPId(k.getId());k.setId("");})
                                .collect(Collectors.toList());
                        keys.addAll(keyList);
                    }
                }
            }
        }
        // 保存字段版本数据
        long count = fields.stream().filter(f -> f.getId().equals("")).count();
        boolean b = dapDataModelTableFieldService.saveBatch(fields);
        // 为hive数据源
        if(databaseType.equals(DatabaseType.HIVE)){
            // 保存分区版本信息
            return partitions.isEmpty() ? b : dapDataModelTablePartitionService.saveBatch(partitions);
        }else {
            // 为pg和mysql数据源
            if(!keys.isEmpty()){
                keys.forEach(key -> {
                    List<String> strings = Arrays.asList(key.getFieldIds().split(","));
                    List<String> ids = Lists.newArrayList();
                    strings.forEach(s -> {
                        String id = fields.stream().filter(f -> s.equals(f.getPId())).map(DapDataModelTableField::getId).collect(Collectors.joining());
                        ids.add(id);
                    });
                    String fieldIds = ids.stream().collect(Collectors.joining(","));
                    key.setFieldIds(fieldIds);
                });
                return dapDataModelSubordinateKeyService.saveBatch(keys);
            }
            return b;
        }
    }

    /**
     * 查询索引或者分区
     * @param list
     * @param databaseType
     * @return
     */
    public List<DapDataModelTableDto> getKeyAndPartitionList(List<DapDataModelTableDto> list, DatabaseType databaseType, String modelId, boolean isNew){
        // hive数据库
        if(databaseType.equals(DatabaseType.HIVE)){
            // 查询该模型的最新的分区信息
            List<DapDataModelTablePartition> partitions = dapDataModelTablePartitionService.getBaseMapper()
                    .selectList(new LambdaQueryWrapper<DapDataModelTablePartition>()
                            .eq(DapDataModelTablePartition::getDataModelId, modelId)
                            .eq(DapDataModelTablePartition::getBmStatus, BmStatus.ON)
                            .eq(isNew ? DapDataModelTablePartition::getPId : DapDataModelTablePartition::getDataNewest, "0"));
            // 进行筛选
            list.parallelStream().forEach(t -> {
                List<DapDataModelTablePartition> collect = partitions.stream()
                        .filter(p -> p.getModelTableId().equals(t.getTableId()))
                        .collect(Collectors.toList());
                t.setPartitionList(collect);
            });
        }else{
            // 查询该模型下的索引信息
            List<DapDataModelSubordinateKey> dapDataModelSubordinateKeys = dapDataModelSubordinateKeyService
                    .getBaseMapper().selectList(new LambdaQueryWrapper<DapDataModelSubordinateKey>()
                            .eq(DapDataModelSubordinateKey::getDataModelId, modelId)
                            .eq(isNew ? DapDataModelSubordinateKey::getPId : DapDataModelSubordinateKey::getDataNewest, "0")
                            .eq(DapDataModelSubordinateKey::getBmStatus, BmStatus.ON));
            // 进行筛选
            list.parallelStream().forEach(t -> {
                List<DapDataModelSubordinateKey> collect = dapDataModelSubordinateKeys.stream().
                        filter(d -> d.getModelTableId().equals(t.getTableId()))
                        .collect(Collectors.toList());
                t.setKeyList(collect);
            });
        }
        return list;
    }

    /**
     * 获取模型版本历史记录列表
     * @param vo
     * @return
     */
    @Override
    public IPage<DapDataModelRecord> getVersionRecord(SearchModelVersionVo vo) {

        Page<DapDataModelRecord> page = this.getBaseMapper().selectModelVersionList(new Page<>(vo.getPageNum(), vo.getPageSize()), vo);
        return page;
    }
}